package com.mai.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.mai.realtime.app.BaseAppV2;
import com.mai.realtime.bean.TradeOrderProvinceBean;
import com.mai.realtime.bean.TrafficSourcePageViewBean;
import com.mai.realtime.util.AtguiguUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.time.Duration;
import java.util.HashMap;

import static com.mai.realtime.common.Constant.*;

/**
 * com.mai.realtime.app.dws.Dws_01_DwsTrafficSourceKeywordPageViewWindow
 * 各渠道流量统计
 */
public class DwsTrafficSourceVcChArPageViewWindow extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsTrafficSourceVcChArPageViewWindow().init(4001,
                3,
                "Dws_01_DwsTrafficSourceKeywordPageViewWindow",
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_DWD_TRAFFIC_UV_DETAIL,
                TOPIC_DWD_TRAFFIC_UJ_DETAIL
        );


    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          HashMap<String, DataStreamSource<String>> streams) {
        // 1.从三条流中读取数据，转换同一个类型union成一条流
        DataStream<TrafficSourcePageViewBean> beanStream = parseAndUnionOne(streams);
        // 2.k开窗聚合
        SingleOutputStreamOperator<TrafficSourcePageViewBean> resultStream = windowAndAgg(beanStream);
        resultStream.print();
        // 3:写出到doris(clickhouse)中
        //writeToDoris(resultStream);
        writeToClickHouse(resultStream);


    }

    private void writeToClickHouse(SingleOutputStreamOperator<TrafficSourcePageViewBean> resultStream) {
        resultStream.addSink(
                new RichSinkFunction<TrafficSourcePageViewBean>() {
                    private Connection conn;
                    private PreparedStatement insertStmt;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = DriverManager.getConnection(
                                "jdbc:clickhouse://hadoop162:8123/default"
                        );
                        insertStmt = conn.prepareStatement("INSERT INTO dws_traffic_vc_ch_ar_source_page_view_window (stt,edt,vc,ch,ar,sc,curDate," +
                                "uvCt,svCt,pvCt,durSum,ujCt,ts) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)");
                    }

                    @Override
                    public void close() throws Exception {
                        insertStmt.close();
                        conn.close();
                    }

                    @Override
                    public void invoke(TrafficSourcePageViewBean value, Context context) throws Exception {
                        insertStmt.setString(1, value.getStt());
                        insertStmt.setString(2, value.getEdt());
                        insertStmt.setString(3, value.getVc());
                        insertStmt.setString(4, value.getCh());
                        insertStmt.setString(4, value.getAr());
                        insertStmt.setString(4, value.getSc());
                        insertStmt.setString(4, value.getCurDate());
                        insertStmt.setString(5, String.valueOf(value.getUvCt()));
                        insertStmt.setString(6, String.valueOf(value.getSvCt()));
                        insertStmt.setString(7, String.valueOf(value.getPvCt()));
                        insertStmt.setString(7, String.valueOf(value.getDurSum()));
                        insertStmt.setString(7, String.valueOf(value.getUjCt()));
                        insertStmt.setString(7, String.valueOf(value.getTs()));
                        insertStmt.execute();
                    }
                }
        );
    }

    private void writeToDoris(SingleOutputStreamOperator<TrafficSourcePageViewBean> resultStream) {
        resultStream
                .map(bean -> {
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;  // 转成json的时候, 属性名使用下划线
//                    config.propertyNamingStrategy = PropertyNamingStrategy.KebabCase;  // 转成json的时候, 属性名使用下划线
                    String json = JSON.toJSONString(bean, config);
                    System.out.println(json);
                    return json;
                })
                .print();
        //.addSink(FlinkSinkUtil.getDorisSink("gmall2022.dws_traffic_vc_ch_ar_is_new_page_view_window"));
        //.addSink(FlinkSinkUtil.getClickhouse("gmall."))
    }

    private SingleOutputStreamOperator<TrafficSourcePageViewBean> windowAndAgg(DataStream<TrafficSourcePageViewBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficSourcePageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getVc() + "_" + bean.getCh() + "_" + bean.getAr() + "_" + bean.getSc())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<TrafficSourcePageViewBean>("late") {
                })
                .reduce(
                        new ReduceFunction<TrafficSourcePageViewBean>() {
                            @Override
                            public TrafficSourcePageViewBean reduce(TrafficSourcePageViewBean bean1,
                                                                    TrafficSourcePageViewBean bean2) throws Exception {
                                bean1.setPvCt(bean1.getPvCt() + bean2.getPvCt());
                                bean1.setSvCt(bean1.getSvCt() + bean2.getSvCt());
                                bean1.setUvCt(bean1.getUvCt() + bean2.getUvCt());
                                bean1.setUjCt(bean1.getUjCt() + bean2.getUjCt());
                                bean1.setDurSum(bean1.getDurSum() + bean2.getDurSum());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TrafficSourcePageViewBean, TrafficSourcePageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TrafficSourcePageViewBean> elements, // 有且仅有一个值: 聚合函数中最终一的结果
                                                Collector<TrafficSourcePageViewBean> out) throws Exception {
                                TrafficSourcePageViewBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis())); // 统计日期

                                out.collect(bean);
                            }
                        }
                );
    }

    private DataStream<TrafficSourcePageViewBean> parseAndUnionOne(HashMap<String,
            DataStreamSource<String>> streams) {

        SingleOutputStreamOperator<TrafficSourcePageViewBean> pvSvDuringSumStream = streams
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .map(json -> {
                    // 计算 pv sv dursum
                    JSONObject obj = JSON.parseObject(json);

                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String sc = common.getString("sc");

                    Long uvCt = 0L;
                    Long svCt = page.getString("last_page_id") == null ? 1L : 0L;
                    Long pvCt = 1L;
                    Long durSum = page.getLong("during_time");
                    Long ujCt = 0L;

                    Long ts = obj.getLong("ts");


                    return new TrafficSourcePageViewBean("", "", // stt edt 只能开窗聚合后才有窗口时间
                            vc, ch, ar, sc,
                            "",  // 当天日期   等待聚合后, 再添加也不迟
                            uvCt, svCt, pvCt, durSum, ujCt,  // 指标列
                            ts // 时间戳
                    );

                });

        // 2. uv
        SingleOutputStreamOperator<TrafficSourcePageViewBean> uvStream = streams
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(json -> {
                    // 计算 pv sv dursum
                    JSONObject obj = JSON.parseObject(json);

                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String sc = common.getString("sc");

                    Long uvCt = 1L;
                    Long svCt = 0L;
                    Long pvCt = 0L;
                    Long durSum = 0L;
                    Long ujCt = 0L;

                    Long ts = obj.getLong("ts");


                    return new TrafficSourcePageViewBean("", "", // stt edt 只能开窗聚合后才有窗口时间
                            vc, ch, ar, sc,
                            "",  // 当天日期   等待聚合后, 再添加也不迟
                            uvCt, svCt, pvCt, durSum, ujCt,  // 指标列
                            ts // 时间戳
                    );

                });

        // 2. uj
        SingleOutputStreamOperator<TrafficSourcePageViewBean> ujStream = streams
                .get(TOPIC_DWD_TRAFFIC_UJ_DETAIL)
                .map(json -> {
                    // 计算 pv sv dursum
                    JSONObject obj = JSON.parseObject(json);

                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String sc = common.getString("sc");

                    Long uvCt = 0L;
                    Long svCt = 0L;
                    Long pvCt = 0L;
                    Long durSum = 0L;
                    Long ujCt = 1L;

                    Long ts = obj.getLong("ts");


                    return new TrafficSourcePageViewBean("", "", // stt edt 只能开窗聚合后才有窗口时间
                            vc, ch, ar, sc,
                            "",  // 当天日期   等待聚合后, 再添加也不迟
                            uvCt, svCt, pvCt, durSum, ujCt,  // 指标列
                            ts // 时间戳
                    );

                });

        return pvSvDuringSumStream.union(uvStream, ujStream);

    }
}
/*
 1:TOPIC_DWD_TRAFFIC_PAGE主题获取 会话总数， 页面总数， 会话总时长
 可以计算的指标：
 会话总数
 会话平均浏览页面数=页面总数/会话总数
 会话平均停留时长=会话总时长/会话总数

  2：DWS只需要计算出：会话总数 页面总数  浏览总时长
  ADS层再计算
  会话平均浏览页面数=页面总数/会话总数
  会话平均停留时长=会话总时长/会话总数
 1. pv sv during_sum
 */
